package com.wande.dataplatform.etl.node;

import cn.hutool.core.util.StrUtil;
import com.wande.dataplatform.common.utils.DataSourceUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 输入节点 - 从数据源读取数据
 *
 * @author dataplatform
 */
@Slf4j
@Component
public class InputNode implements EtlNode {

    @Override
    public EtlNodeResult execute(EtlContext context) {
        try {
            Map<String, Object> config = context.getConfig();
            
            // 获取配置
            String jdbcUrl = (String) config.get("jdbcUrl");
            String username = (String) config.get("username");
            String password = (String) config.get("password");
            String tableName = (String) config.get("tableName");
            String whereClause = (String) config.get("whereClause");
            
            // 构建SQL
            String sql = "SELECT * FROM " + tableName;
            if (StrUtil.isNotBlank(whereClause)) {
                sql += " WHERE " + whereClause;
            }
            
            log.info("执行输入节点SQL: {}", sql);
            
            // 执行查询
            List<Map<String, Object>> data = new ArrayList<>();
            try (Connection conn = DataSourceUtils.getConnection(jdbcUrl, username, password);
                 PreparedStatement stmt = conn.prepareStatement(sql);
                 ResultSet rs = stmt.executeQuery()) {
                
                ResultSetMetaData metaData = rs.getMetaData();
                int columnCount = metaData.getColumnCount();
                
                while (rs.next()) {
                    Map<String, Object> row = new HashMap<>();
                    for (int i = 1; i <= columnCount; i++) {
                        String columnName = metaData.getColumnName(i);
                        Object value = rs.getObject(i);
                        row.put(columnName, value);
                    }
                    data.add(row);
                }
            }
            
            log.info("输入节点读取数据: {} 行", data.size());
            context.getStatistics().addSourceRows(data.size());
            
            return EtlNodeResult.success(data);
            
        } catch (Exception e) {
            log.error("输入节点执行失败", e);
            return EtlNodeResult.failure("输入节点执行失败: " + e.getMessage());
        }
    }

    @Override
    public String getNodeType() {
        return INPUT;
    }

    @Override
    public boolean validate(Map<String, Object> config) {
        return config.containsKey("jdbcUrl") 
            && config.containsKey("username")
            && config.containsKey("tableName");
    }
}
